package org.shj.spark.operator;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

public class AggregateOperator {

	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setMaster("local").setAppName("wc");
		JavaSparkContext ctx = new JavaSparkContext(conf);
		JavaRDD<String> text = ctx.textFile("test.txt");//每一行的数据是 String
		
		JavaRDD<String> words = text.flatMap(new FlatMapFunction<String,String>(){
			private static final long serialVersionUID = -419305824284261543L;

			public Iterator<String> call(String line) throws Exception {
				return Arrays.asList(line.split(" ")).iterator();
			}
		});
		
		JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>(){
			private static final long serialVersionUID = -7788686310085294474L;

			public Tuple2<String, Integer> call(String word) throws Exception {
				return new Tuple2<String, Integer>(word, 1);
			}
		});
		
		//aggregateByKey其实和reduceByKey差不多，reduceByKey是aggregateByKey的简化版
		//aggregateByKey里面的参数需要三个
		//第一个：每个Key的初始值 
		//第二个， Seq Function，进行shuffle时， map端的操作
		//第三个，进行shuffle时 reduce端的操作
		JavaPairRDD<String, Integer> wordCount = pairs.aggregateByKey(0, 
				new Function2<Integer, Integer, Integer>(){
					private static final long serialVersionUID = 1L;
		
					public Integer call(Integer v1, Integer v2) throws Exception {
						return v1 + v2;
					}
			
				}, 
				new Function2<Integer, Integer, Integer>(){
					private static final long serialVersionUID = 1L;
		
					public Integer call(Integer v1, Integer v2) throws Exception {
						return v1 + v2;
					}
					
				});
		
		List<Tuple2<String, Integer>> list = wordCount.collect();
		for(Tuple2<String, Integer> wc : list){
			System.out.println(wc);
		}
		
		ctx.close();

	}

}
